package cn.da.shuai.cool.mq.mqtt.config;

import cn.da.shuai.cool.mq.mqtt.constant.MqttConstant;
import jakarta.annotation.Resource;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@Slf4j
@Configuration
@EnableIntegration
public class MqttConfig {

    @Setter(onMethod_ = @Resource)
    private MqttProperties mqttProperties;

    @Bean
    public MqttPahoClientFactory getFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttProperties.getFinalOptions());
        return factory;
    }

    @Bean(name = MqttConstant.OUT_CHANNEL)
    public MessageChannel outChannel() {
        return new DirectChannel();
    }

    @Bean(name = MqttConstant.IN_CHANNEL)
    public MessageChannel inChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = MqttConstant.OUT_CHANNEL)
    public MessageHandler outHandler() {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(
                mqttProperties.getClientId(), getFactory()
        );
        handler.setAsync(true);
        handler.setDefaultTopic(mqttProperties.getTopic());
        log.info("send message handler status: {}", handler);
        return handler;
    }

    @Bean
    @ServiceActivator(inputChannel = MqttConstant.IN_CHANNEL)
    public MessageHandler inHandler() {
        return message -> log.info("in handler: {}", message.getPayload());
    }

    @Bean
    public MessageProducer getAdapter() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperties.getClientId(),
                getFactory(),
                mqttProperties.getTopic()
        );
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setOutputChannel(inChannel());
        log.info("subscribe adapter status: {}", adapter);
        return adapter;
    }

}
